package com.huawei.bigdata.kafka.controller;

import com.huawei.bigdata.kafka.example.security.LoginUtil;
import com.huawei.bigdata.kafka.service.FileReadService;
import com.huawei.bigdata.kafka.temp.TestConsumer;
import com.huawei.bigdata.kafka.temp.TestProducer;
import jakarta.websocket.server.PathParam;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.io.IOException;

@RestController
@RequestMapping("/test")
@Slf4j
public class TestController {
    @Value("${conf.kafka.test.user-keytab-file}")
    private String userKeytabFile;
    @Value("${conf.kafka.test.user-principal}")
    private String userPrincipal;
    @Autowired
    private TestProducer testProducer;

    @Autowired
    private TestConsumer testConsumer;

    @Autowired
    private FileReadService fileReadService;
    /**
     * 发送消息
     * @param topic
     * @param isAsync  = 1 异步发送，= 其它 同步发送
     */
    @GetMapping("/send/{topic}/{isAsync}")
    void send(@PathVariable String topic, @PathVariable("isAsync") Integer isAsync) {
        log.info("send:{}:{}", topic, isAsync);
        testProducer.send(topic, "test-key", "test-value", isAsync);
    }

    /**
     * 接收消息
     * @param topic
     */
    @GetMapping("/get/{topic}")
    void get(@PathVariable String topic) {
        log.info("get:{}", topic);
        testConsumer.consume(topic);
    }

    /**
     * 停止接收
     */
    @GetMapping("/close")
    void close() {
        log.info("close");
        testConsumer.closeThread();
    }

    /**
     * 停止接收
     */
    @GetMapping("/msg")
    void msg() {
        log.info("msg");
        fileReadService.sendMsg();
    }

}
